package com.study.flink.sink

import java.util

import org.apache.flink.api.common.functions.RuntimeContext
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.connectors.elasticsearch.{ElasticsearchSinkFunction, RequestIndexer}
import org.apache.flink.streaming.connectors.elasticsearch6.ElasticsearchSink
import org.apache.http.HttpHost
import org.elasticsearch.client.Requests

/**
  * Elasticsearch Sink Demo
  *
  * @author stephen
  * @date 2019-07-22 15:10
  */
object ScalaElasticsearchSinkDemo {

  def main(args: Array[String]): Unit = {
    // 1 获取执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 2 获取数据
    import org.apache.flink.api.scala._
    val dataStream = env.socketTextStream("localhost", 9999)

    // 3
    val wordStream = dataStream.flatMap(_.split(" "))

    // 4 输出数据到ES
    val httpHosts = new util.ArrayList[HttpHost]()
    httpHosts.add(new HttpHost("localhost", 9200,"http"))
    val esSink = new ElasticsearchSink.Builder[String](httpHosts, new ElasticsearchSinkFunction[String] {
      override def process(t: String, runtimeContext: RuntimeContext, requestIndexer: RequestIndexer): Unit = {
        val json = new util.HashMap[String, Int]()
        json.put(t, 1)
        val indexRequest = Requests.indexRequest().index("test_idx").`type`("word").source(json)
        requestIndexer.add(indexRequest)
      }
    }).build()
    wordStream.addSink(esSink)

    // 5 启动任务
    env.execute("Elasticsearch Sink Demo")

  }

}
